package com.aboo.frame.util;

import com.aboo.frame.config.zookeeper.ZkWatcher;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Slf4j
public class ZkClient {

    private static ZooKeeper zkclient;
    private static String zkServerCluster;
    private static int zkTimeout;

    public static ZooKeeper connect() throws IOException, InterruptedException {
        CountDownLatch cdl = new CountDownLatch(1);
        log.info("准备建立zk服务");
        zkclient = new ZooKeeper(zkServerCluster, zkTimeout, new ZkWatcher(cdl, "建立连接"));
        log.info("完成建立zk服务");
        //这里为了等待wather监听事件结束
        cdl.await(10L, TimeUnit.SECONDS);
        return zkclient;
    }

    public static ZooKeeper reconnect() throws IOException, InterruptedException {
        CountDownLatch cdl = new CountDownLatch(1);
        log.info("准备重新连接zk服务");
        /**恢复会话连接**/
        long sessionId = zkclient.getSessionId();
        byte[] sessionPwd = zkclient.getSessionPasswd();
        zkclient = new ZooKeeper(zkServerCluster, zkTimeout, new ZkWatcher(cdl, "重新连接"), sessionId, sessionPwd);
        log.info("完成重新连接zk服务");
        //这里为了等待wather监听事件结束
        cdl.await();
        return zkclient;
    }

    public static void create(String nodePath, String nodeData) throws KeeperException, InterruptedException {
        log.info("开始创建节点：{}， 数据：{}", nodePath, nodeData);
        List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
        CreateMode createMode = CreateMode.PERSISTENT;
        String result = zkclient.create(nodePath, nodeData.getBytes(), acl, createMode);
        //创建节点有两种，上面是第一种，还有一种可以使用回调函数及参数传递，与上面方法名称相同。
        log.info("创建节点返回结果：{}", result);
        log.info("完成创建节点：{}， 数据：{}", nodePath, nodeData);
    }

    public static Stat queryStat(String nodePath) throws KeeperException, InterruptedException {
        log.info("准备查询节点Stat，path：{}", nodePath);
        Stat stat = zkclient.exists(nodePath, false);
        log.info("结束查询节点Stat，path：{}，version：{}", nodePath, stat.getVersion());
        return stat;
    }

    public static String queryData(String nodePath) throws KeeperException, InterruptedException {
        log.info("准备查询节点Data,path：{}", nodePath);
        byte[] bytes = zkclient.getData(nodePath, false, queryStat(nodePath));
        String data = new String(bytes);
        log.info("结束查询节点Data,path：{}，Data：{}", nodePath, data);
        return data;
    }

    public static Stat update(String nodePath, String nodeData) throws KeeperException, InterruptedException {
        //修改节点前先查询该节点信息
        Stat stat = queryStat(nodePath);
        log.info("准备修改节点，path：{}，data：{}，原version：{}", nodePath, nodeData, stat.getVersion());
        Stat newStat = zkclient.setData(nodePath, nodeData.getBytes(), stat.getVersion());
        log.info("完成修改节点，path：{}，data：{}，现version：{}", nodePath, nodeData, newStat.getVersion());
        return stat;
    }

    public static void delete(String nodePath) throws InterruptedException, KeeperException {
        //删除节点前先查询该节点信息
        Stat stat = queryStat(nodePath);
        log.info("准备删除节点，path：{}，原version：{}", nodePath, stat.getVersion());
        zkclient.delete(nodePath, stat.getVersion());
        log.info("完成删除节点，path：{}", nodePath);
    }

    public static void setZkclient(ZooKeeper zkclient) {
        ZkClient.zkclient = zkclient;
    }

    public static void setZkServerCluster(String zkServerCluster) {
        ZkClient.zkServerCluster = zkServerCluster;
    }

    public static void setZkTimeout(int zkTimeout) {
        ZkClient.zkTimeout = zkTimeout;
    }

}
